package utils;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.Scanner;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DefaultStringifier;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.TextOutputFormat;

import regression.model.AbstractRegression;
import utils.StringUtil;

public class HadoopUtil {
    /**
     * Save model on HDFS
     * @param job
     * @param model
     * @param modelHDFSPath
     */
    public static void HDFSSaveModel(JobConf job, AbstractRegression model, String modelHDFSPath){
	job.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization,org.apache.hadoop.io.serializer.WritableSerialization");
	DefaultStringifier<AbstractRegression> output = new DefaultStringifier<AbstractRegression>(job, AbstractRegression.class); 
	try {
	    String s = output.toString(model);
	    RecordWriter<Text, Text> rwFile = null;
	    rwFile = new TextOutputFormat<Text, Text>().getRecordWriter(null,
		    job, modelHDFSPath, null);
	    rwFile.write(new Text(StringUtil.getIP()), new Text(s));
	    rwFile.close(null);
	} catch (IOException e) {
	    e.printStackTrace();
	}
    }
    /**
     * Get model from HDFS
     * @param job
     * @param modelHDFSPath
     * @return abstract regression model
     */
    public static AbstractRegression HDFSLoadModel(JobConf job, String modelHDFSPath){
	job.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization,org.apache.hadoop.io.serializer.WritableSerialization");
	DefaultStringifier<AbstractRegression> ds = new DefaultStringifier<AbstractRegression>(job, AbstractRegression.class); 
	try {
	    Scanner modelString = new Scanner(new File(modelHDFSPath));
	    AbstractRegression rs = (AbstractRegression)ds.fromString(modelString.next());  
	    modelString.close();
	    return rs;
	} catch (IOException e) {
	    e.printStackTrace();
	    return null;
	}
    }
    
    /**
     * Send file to HDFS
     * 
     * @param src
     * @param dst
     */
    public static void localFile2HDFS(String src, String dst) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    Path dstPath = new Path(dst);
	    fs.copyFromLocalFile(srcPath, dstPath);
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }

    /**
     * Get file from HDFS
     * 
     * @param delSrc
     * @param src
     * @param dst
     */
    public static void HDFSFile2Local(boolean delSrc, String src, String dst) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    Path dstPath = new Path(dst);
	    fs.copyToLocalFile(delSrc, srcPath, dstPath);
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
    
    /**
     * Write map data
     * @param dst
     * @param resMap
     */
    public static void HDFSWriteMapData(String dst,
	    Map<Integer, double[]> resMap) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path dstPath = new Path(dst);
	    BufferedWriter d = new BufferedWriter(new OutputStreamWriter(fs
		    .create(dstPath)));
	    Iterator<Integer> it = resMap.keySet().iterator();
	    while (it.hasNext()) {
		Integer key = it.next();
		double values[] = resMap.get(key);
		for (int i = 0; i < values.length - 1; i++)
		    d.write(values[i] + "\t");
		d.write("\n");
	    }
	    d.flush();
	    d.close();
	} catch (Exception e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
    /**
     * Read array data
     * @param src
     * @return
     */
    public static double[][] HDFSReadArrayData(String src) {
	ArrayList<ArrayList<Double>> array = new ArrayList<ArrayList<Double>>();
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);

	    BufferedReader d = new BufferedReader(new InputStreamReader(fs
		    .open(srcPath)));
	    while (d.ready()) {
		String line = d.readLine();
		String items[] = line.split("\t");
		ArrayList<Double> values = new ArrayList<Double>();
		for (int i = 1; i < items.length; i++) {
		    values.add(Double.parseDouble(items[i]));
		}
	    }
	    d.close();

	    double res[][] = new double[array.size()][array.get(0).size()];
	    for (int i = 0; i < array.size(); i++)
		for (int j = 0; j < array.get(i).size(); j++)
		    res[i][j] = array.get(i).get(j);
	    return res;
	    
	} catch (Exception e) {
	    e.printStackTrace();
	    return null;
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
    /**
     * Read map data from one fold
     * @param src
     * @return
     */
    @SuppressWarnings("deprecation")
    public static Map<Integer, double[]> HDFSReadMapData(String src) {
	Map<Integer, double[]> resMap = new TreeMap<Integer, double[]>();
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    if (fs.isDirectory(srcPath)) {
		FileStatus subFiles[] = fs.listStatus(srcPath);
		for (FileStatus it : subFiles) {
		    BufferedReader d = new BufferedReader(
			    new InputStreamReader(fs.open(it.getPath())));
		    while (d.ready()) {
			String line = d.readLine();
			String kv[] = line.split("\t");
			String items[] = kv[1].split(",");
			double values[] = new double[items.length];
			Integer key = Integer.parseInt(kv[0]);
			if (resMap.containsKey(key))
			    values = resMap.get(key);
			for (int i = 0; i < items.length; i++) {
			    values[i] += Double.parseDouble(items[i]);
			}
			resMap.put(key, values);
		    }
		    d.close();
		}
	    }
	    Iterator<Integer> it = resMap.keySet().iterator();
	    while (it.hasNext()) {
		Integer key = it.next();
		double values[] = resMap.get(key);
		for (int i = 0; i < values.length - 1; i++)
		    values[i] /= values[values.length - 1];
	    }
	    return resMap;
	} catch (Exception e) {
	    e.printStackTrace();
	    return null;
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
    /**
     * Move files on HDFS
     * @param delSrc
     * @param src
     * @param dst
     */
    public static void HDFSMove(boolean delSrc, String src, String dst) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    FileStatus fsta[] = fs.globStatus(srcPath);
	    for (FileStatus it : fsta) {
		Path singlePath = it.getPath();
		Path dstPath = new Path(dst);
		FileUtil.copyMerge(fs, singlePath, fs, dstPath, delSrc, conf,
			"");
	    }
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
    /**
     * Delete file on the HDFS
     * @param src
     */
    public static void HDFSDelete(String src) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    fs.delete(srcPath, true);
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
}
